package com.radon.kafka.conf;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;


/**
 * 消费者需要在接收的方法上添加@KafkaListener，用于监听对应的topic,可以配置topic多个
 */
@Component
public class KafKaConsumer {

    private final Logger logger = LoggerFactory.getLogger(KafKaConsumer.class);


    /**
     * @Description: 可以同时订阅多主题，只需按数组格式即可，也就是用“，”隔开
     * @Author:杨攀
     * @Since: 2019年7月10日上午11:26:16
     * @param record
     */
    /*@KafkaListener(topics = { Constant.TEST_KAFKA_TOPIC })
    public void receive(ConsumerRecord<?, ?> record) {

        logger.info("消费得到的消息---key: " + record.key());
        logger.info("消费得到的消息---value: " + record.value().toString());

    }*/

    /**
     * @Description: 可以同时订阅多主题，只需按数组格式即可，也就是用“，”隔开
     * @Author:杨攀
     * @Since: 2019年7月10日上午11:26:16
     * @param record
     */
    @KafkaListener(topics = { Constant.TEST_KAFKA_TOPIC }, containerFactory = "kafkaListenerContainerFactory")
    public void receive2(ConsumerRecord<?, ?> record) {

        logger.info("消费者1分组1");
        logger.info("消费得到的消息---key: " + record.key());
        logger.info("消费得到的消息---value: " + record.value().toString());

    }

    /**
     * @Description: 可以同时订阅多主题，只需按数组格式即可，也就是用“，”隔开
     * @Author:杨攀
     * @Since: 2019年7月10日上午11:26:16
     * @param record
     */
    @KafkaListener(topics = { Constant.TEST_KAFKA_TOPIC }, containerFactory = "kafkaListenerContainerFactory2")
    public void receive3(ConsumerRecord<?, ?> record) {

        logger.info("消费者2分组1");
        logger.info("消费得到的消息---key: " + record.key());
        logger.info("消费得到的消息---value: " + record.value().toString());

    }

    /**
     * @Description: 可以同时订阅多主题，只需按数组格式即可，也就是用“，”隔开
     * @Author:杨攀
     * @Since: 2019年7月10日上午11:26:16
     * @param record
     */
    @KafkaListener(topics = { Constant.TEST_KAFKA_TOPIC }, containerFactory = "kafkaListenerContainerFactory3")
    public void receive4(ConsumerRecord<?, ?> record) {

        logger.info("消费者3分组2");
        logger.info("消费得到的消息---key: " + record.key());
        logger.info("消费得到的消息---value: " + record.value().toString());

    }

}
